package com.yangziqiang.kafka;

import com.alibaba.fastjson.JSON;
import com.yangziqiang.entity.EsMovies;
import com.yangziqiang.mapper.MoviesRepository;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * @ClassName MoviesCustomer
 * @Description TODO
 * @Author 18516
 * @DATE 2022/2/9 09:13:44
 * @Version 1.0
 */
@Component
public class MoviesConsumer {

    @Autowired
    RedisTemplate redisTemplate;

    @Autowired
    MoviesRepository moviesRepository;

    @KafkaListener(topics = {"test111"})
    public void moviesConsumer(ConsumerRecord<String, String> consumerRecord, Acknowledgment acknowledgment) {
        String key = consumerRecord.key();
        String value = consumerRecord.value();
        String substring = key.substring(0, 3);
        if ("add".equals(substring)) {
            if (redisTemplate.opsForValue().setIfAbsent("addMovies:" + key, value, 10, TimeUnit.HOURS)) {
                return;
            }
            EsMovies esMovies = JSON.parseObject(value, EsMovies.class);
            moviesRepository.save(esMovies);
            acknowledgment.acknowledge();
            redisTemplate.delete("addMovies:" + key);
        }
        if ("upd".equals(substring)) {
            if (redisTemplate.opsForValue().setIfAbsent("updMovies:" + key, value, 10, TimeUnit.HOURS)) {
                return;
            }
            EsMovies esMovies = JSON.parseObject(value, EsMovies.class);
            moviesRepository.save(esMovies);
            acknowledgment.acknowledge();

            redisTemplate.delete("updMovies:" + key);
        }
        if ("cha".equals(substring)) {
            if (redisTemplate.opsForValue().setIfAbsent("chaMovies:" + key, value, 10, TimeUnit.HOURS)) {
                return;
            }
            EsMovies esMovies = JSON.parseObject(value, EsMovies.class);
            moviesRepository.save(esMovies);
            acknowledgment.acknowledge();
            redisTemplate.delete("chaMovies:" + key);
        }
        if ("del".equals(substring)) {
            if (redisTemplate.opsForValue().setIfAbsent("delMovies:" + key, value, 10, TimeUnit.HOURS)) {
                return;
            }
            Integer mid = JSON.parseObject(value, Integer.class);
            moviesRepository.deleteById(mid);
            acknowledgment.acknowledge();
            redisTemplate.delete("delMovies:" + key);
        }

    }
}
